
package com.kalvan.core.mq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.LinkedList;
import java.util.List;

/**
 * 用于创建消费者
 *
 * @author chenliang
 */
@Slf4j
@Component
public class MqConsumer {

    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;

    private List<DefaultMQPushConsumer> consumerContainer = new LinkedList<>();

    @PostConstruct
    public void start() {
        //代收异步通知并发消费(控制线程池大小)
        TransNotifyDsConsumer mqConsumer = new TransNotifyDsConsumer(50, 100, 20, 100);
        consumerContainer.add(createConsumer("test-topic", mqConsumer));
    }


    /**
     * 创建一个并发消息消费者
     */
    public DefaultMQPushConsumer createConsumer(String topic, MessageListenerConcurrently messageListener) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Group_" + topic);
        log.info("MQ：启动消费者:{}，MQ地址：{}，消费主题：{}", this.getClass(), namesrvAddr, topic);
        try {
            consumer.setNamesrvAddr(namesrvAddr);
            //实例名默认是ip+进程号
            //consumer.setInstanceName("PushConsumer1");
            // 从消息队列头开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            // 订阅主题
            consumer.subscribe(topic, "*");
            // 注册消息监听器
            consumer.registerMessageListener(messageListener);
            // 启动消费端
            consumer.start();
        } catch (MQClientException e) {
            log.error("MQ：消费者{} 异常信息：{}-{}", this.getClass(), e.getResponseCode(), e.getErrorMessage());
            throw new RuntimeException(e.getMessage(), e);
        }
        return consumer;
    }


    @PreDestroy
    public void destroy() {
        if (!CollectionUtils.isEmpty(consumerContainer)) {
            for (DefaultMQPushConsumer consumer : consumerContainer) {
                log.info("MQ：关闭消费者{}", consumer.getConsumerGroup());
                consumer.shutdown();
            }
        }
    }
}
